package cn.itcast.streaming.source;

import cn.itcast.streaming.bean.ElectricFenceRuleTmp;
import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.JDBCUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

//读取电子围栏规则表数据和电子围栏关联的车辆数据
//一个车辆可能对应多个电子围栏规则，所以返回的数据可能是有多个规则数据
//当前业务中同一个车辆只能关联一个生效的电子围栏，所以车辆匹配的电子围栏规则是一对一的
public class MysqlElectricFenceSource extends RichSourceFunction<HashMap<String, ElectricFenceRuleTmp>> {

    private static Logger logger = LoggerFactory.getLogger(MysqlElectricFenceSource.class);
    private ParameterTool parameterTool; //全局参数对象
    private Connection conn;
    private Statement statement;
    private Boolean isRunning = true;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        conn = JDBCUtil.getConnection();
        statement = conn.createStatement();
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(statement!=null && !statement.isClosed()) statement.close();
        if(conn!=null && !conn.isClosed()) conn.close();
    }

    @Override
    public void run(SourceContext<HashMap<String, ElectricFenceRuleTmp>> sourceContext) throws Exception {
        while (isRunning){
            String sql = "select vin,name,address,radius,longitude,latitude,start_time,end_time,setting_id\n" +
                    "from vehicle_networking.electronic_fence_vins vin\n" +
                    "inner join vehicle_networking.electronic_fence_setting setting\n" +
                    "on vin.setting_id=setting.id\n" +
                    "and setting.status=1;";
            ResultSet resultSet = statement.executeQuery(sql);
            HashMap<String, ElectricFenceRuleTmp> map = new HashMap<>();
            while (resultSet.next()){
                int setting_id = resultSet.getInt("setting_id");
                String name = resultSet.getString("name");
                String address = resultSet.getString("address");
                Float radius = resultSet.getFloat("radius");
                double longitude = resultSet.getDouble("longitude");
                double latitude = resultSet.getDouble("latitude");
                Date start_time = DateUtil.convertStringToDate2(resultSet.getString("start_time"));
                Date end_time = DateUtil.convertStringToDate2(resultSet.getString("end_time"));
                ElectricFenceRuleTmp electricFenceResultTmp = new ElectricFenceRuleTmp(
                        setting_id, name, address, radius, longitude, latitude, start_time, end_time);
                map.put(resultSet.getString("vin"),electricFenceResultTmp);
            }
            if(map.isEmpty()){
                logger.info("没有获取到电子围栏规则数据....");
            }else {
                logger.info("获取的规则数据为："+map);
            }
            resultSet.close();
            sourceContext.collect(map);
            TimeUnit.SECONDS.sleep(Long.parseLong(parameterTool.getRequired("vehinfo.millionseconds")));
        }
    }


    @Override//取消数据读取
    public void cancel() {
        isRunning = false;
    }
}
